package com.aotain.jupiter.domain;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.common.CommonFunction;
import com.aotain.jupiter.common.URLStatTuple;

public class DomainMap  implements MapFunction<Tuple2<String,String>,Tuple2<String,URLStatTuple>>{

	/**
	 * 
	 */
	private static final long serialVersionUID = -8618893232829700041L;

	@Override
	public Tuple2<String, URLStatTuple> map(Tuple2<String, String> value)
			throws Exception {
		// TODO Auto-generated method stub
//		//dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" 
//		//+ domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" 
//		// +upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" 
//		// +dnStreamOctets+ "|" +dnStreamPacket+ "|" +refer
		
		Tuple2<String, URLStatTuple> ret = new Tuple2<String, URLStatTuple>();
		String logString = value.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		if(items.length <= 10)
			return ret;
//		String sip = value.f0;
//		String dip = items[0];
//		String gis = items[4];
//		String idc = items[9];
		String domainname = items[5];

		if(domainname.isEmpty())
			return ret;
		
		domainname = CommonFunction.getTopDomain(domainname);
		if(domainname.isEmpty())
			return ret;
		
//		long upStreamOctets = 0; //UPSTREAMOCTETS
//		long upStreamPacket = 0; //UPSTREAMPACKET
//		long dnStreamOctets = 0; //DNSTREAMOCTETS
//		long dnStreamPacket = 0; //DNSTREAMPACKET
		
//		String tripleKey = dip + "|" + sip +"|" + url + "|" 
//				+ gis + "|" + idc + "|" + useragent + "|" + cookie;
		
//		upStreamOctets = Long.parseLong(items[10]);
//		upStreamPacket = Long.parseLong(items[11]);
//		dnStreamOctets = Long.parseLong(items[15]);
//		dnStreamPacket = Long.parseLong(items[16]);
		
		URLStatTuple info = new URLStatTuple();
		
		info.setHTTPNum(1l);
//		String retKey = String.format("%s", tripleKey);
		
		ret.setFields(domainname, info);
		return ret;
	}

}
